package org.zhaya.sdk.processors;

import ch.randelshofer.fastdoubleparser.FastDoubleParser;
import com.google.common.collect.Maps;
import io.logz.sawmill.Doc;
import io.logz.sawmill.ProcessResult;
import io.logz.sawmill.Processor;
import io.logz.sawmill.annotations.ProcessorProvider;
import io.logz.sawmill.exceptions.ProcessorConfigurationException;
import io.logz.sawmill.utilities.JsonUtils;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.LocaleUtils;
import org.apache.commons.lang3.StringUtils;

import java.math.BigDecimal;
import java.text.ParseException;
import java.time.*;
import java.time.chrono.IsoChronology;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.format.ResolverStyle;
import java.time.format.SignStyle;
import java.time.temporal.ChronoField;
import java.time.temporal.TemporalAccessor;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static com.google.common.base.Preconditions.checkState;
import static java.time.format.DateTimeFormatter.ISO_LOCAL_DATE;
import static java.time.temporal.ChronoField.*;
import static java.util.Objects.requireNonNull;

/**
 * 扩展DateProcessor,除了转换成ES的时间格式外，增加转换成毫秒或秒的功能。
 * <p>
 * 在官方的基础上增加了一个targetFormat属性，用来指定输出的格式，可以是ES, UNIX,
 * UNIX_MS三种，默认是UNIX_MS,表示转成毫秒值。
 *
 * @author kevin
 */
@ProcessorProvider(type = "date", factory = DateProcessor.Factory.class)
public class DateProcessor implements Processor {

    /**
     * Joda ISODateTimeFormat dateParser description The motivation is supporting
     * logstash ISO8601 pattern generic ISO date parser for parsing dates with a
     * possible zone.
     * <p>
     * It accepts formats described by the following syntax:
     *
     * <pre>
     * date              = date-element ['T' offset]
     * date-element      = std-date-element | ord-date-element | week-date-element
     * std-date-element  = yyyy ['-' MM ['-' dd]]
     * ord-date-element  = yyyy ['-' DDD]
     * week-date-element = xxxx '-W' ww ['-' e]
     * offset            = 'Z' | (('+' | '-') HH [':' mm [':' ss [('.' | ',') SSS]]])
     * </pre>
     */
    public static final DateTimeFormatter iso8601 = new DateTimeFormatterBuilder().parseCaseInsensitive()
            .append(ISO_LOCAL_DATE).appendLiteral('T').appendValue(HOUR_OF_DAY, 2).appendLiteral(':')
            .appendValue(MINUTE_OF_HOUR, 2).optionalStart().appendLiteral(':').appendValue(SECOND_OF_MINUTE, 2)
            .optionalStart().optionalStart().appendLiteral(',').optionalEnd().optionalStart().appendLiteral('.')
            .optionalEnd().appendFraction(NANO_OF_SECOND, 0, 9, false).optionalStart().appendOffsetId().optionalEnd()
            .optionalEnd().optionalEnd().toFormatter().withChronology(IsoChronology.INSTANCE)
            .withResolverStyle(ResolverStyle.STRICT);
    private static Map<String, DateTimeFormatter> dateTimePatternToFormatter = new ConcurrentHashMap<>();
    /**
     * 存放不同的日期模板格式的sdf的Map
     */
    private static Map<String, DateTimeFormatter> sdfMap = Maps.newConcurrentMap();

    static {
        dateTimePatternToFormatter.put("ISO8601", iso8601);
        dateTimePatternToFormatter.put("UNIX", new DateTimeFormatterBuilder()
                .appendValue(ChronoField.INSTANT_SECONDS, 1, 10, SignStyle.NEVER).toFormatter());
        dateTimePatternToFormatter.put("UNIX_MS",
                new DateTimeFormatterBuilder().appendValue(ChronoField.INSTANT_SECONDS, 1, 10, SignStyle.NEVER)
                        .appendValue(ChronoField.MILLI_OF_SECOND, 3).toFormatter());
    }

    private final String field;
    private final String targetField;
    private final List<String> formats;
    private final String locale;
    private final List<DateTimeFormatter> formatters;
    private final ZoneId timeZone;
    private final TargetFormat targetFormat;
    private final DateTimeFormatter outputFormatter;

    public DateProcessor(String field, String targetField, TargetFormat targetFormat, List<String> formats, String locale,
                         ZoneId timeZone) {
        checkState(CollectionUtils.isNotEmpty(formats), "formats cannot be empty");
        this.field = requireNonNull(field, "field cannot be null");
        this.targetField = requireNonNull(targetField, "target field cannot be null");
        this.targetFormat = requireNonNull(targetFormat, "target format cannot be null");
        this.locale = locale;
        this.formats = formats;
        this.timeZone = timeZone;
        this.formatters = new ArrayList<>();
        this.outputFormatter = computeAndGetFormatter(formats.get(0), timeZone);
        formats.forEach(format -> {
            DateTimeFormatter formatter = computeAndGetFormatter(format, timeZone);

            formatters.add(formatter);
        });
    }

    /**
     * 构造格式化
     *
     * @param format
     * @param timeZone
     * @return
     */
    private DateTimeFormatter computeAndGetFormatter(String format, ZoneId timeZone) {
        if (format.toUpperCase().startsWith("UNIX")) {
            return dateTimePatternToFormatter.get(format).withZone(timeZone == null ? ZoneId.systemDefault() : timeZone).withLocale(LocaleUtils.toLocale(StringUtils.defaultIfBlank(locale, Locale.US.toString())));
        }

        try {
            return dateTimePatternToFormatter.computeIfAbsent(format, k -> new DateTimeFormatterBuilder().parseCaseInsensitive().appendPattern(format).toFormatter().withLocale(LocaleUtils.toLocale(StringUtils.defaultIfBlank(locale, Locale.US.toString()))));
        } catch (IllegalArgumentException e) {
            throw new RuntimeException(String.format("failed to create date processor, format [%s] is not valid", format), e);
        }
    }

    @Override
    public ProcessResult process(Doc doc) {
        if (!doc.hasField(field)) {
            return ProcessResult.failure(String.format("failed to process date, field in path [%s] is missing", field));
        }

        Object dateTimeDocValue = doc.getField(field);
        if (dateTimeDocValue != null) {
            try {
                //时间戳转日期字符串
                if (dateTimeDocValue instanceof Number) {
                    long epocTimeInMilis = BigDecimal.valueOf(FastDoubleParser.parseDouble(String.valueOf(dateTimeDocValue)))
                            .multiply(TargetFormat.UNIX.equals(targetFormat) ? BigDecimal.valueOf(1000) : BigDecimal.ONE).longValue();
                    ZonedDateTime dateTime = getEpochMillisDateTime(epocTimeInMilis);
                    doc.addField(targetField, dateTime.format(outputFormatter));
                } else {
                    doc.addField(targetField, getTime(dateTimeDocValue.toString(), locale, formats.get(0), timeZone, targetFormat));
                }
            } catch (Exception e) {
                return ProcessResult
                        .failure(String.format("failed to parse date in path [%s], [%s] is not one of the formats [%s]",
                                field, dateTimeDocValue, formats));
            }
            return ProcessResult.success();
        }

        return ProcessResult.success();
    }

    private ZonedDateTime getEpochMillisDateTime(long unixTimestamp) {
        Instant instant = Instant.ofEpochMilli(unixTimestamp);
        if (timeZone == null) {
            return ZonedDateTime.ofInstant(instant, ZoneId.systemDefault());
        } else {
            return ZonedDateTime.ofInstant(instant, timeZone);
        }
    }

    public static class Factory implements Processor.Factory {

        @Override
        public Processor create(Map<String, Object> config) {
            Configuration dateConfig = JsonUtils.fromJsonMap(Configuration.class, config);

            if (CollectionUtils.isEmpty(dateConfig.getFormats())) {
                throw new ProcessorConfigurationException("cannot create date processor without any format");
            }

            String field = dateConfig.getField();
            String targetField = dateConfig.getTargetField();
            List<String> formats = dateConfig.getFormats();
            String timeZone = dateConfig.getTimeZone();
            TargetFormat targetFormat = dateConfig.getTargetFormat();
            String locale = dateConfig.getLocale();
            ZoneId zoneId = timeZone != null ? ZoneId.of(timeZone) : null;

            return new DateProcessor(field, targetField, targetFormat, formats, locale, zoneId);
        }
    }

    public enum TargetFormat {
        UNIX, UNIX_MS
    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Configuration implements Processor.Configuration {
        private String field;
        private String locale = Locale.US.toString();
        private String targetField = "@timestamp";
        private TargetFormat targetFormat = TargetFormat.UNIX_MS;

        /**
         * The format of the date string. The format in this String is documented in
         * https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html.
         * Example: "yyyy MM dd"
         */
        private List<String> formats;

        /**
         * The time zone The zone in this String is documented in
         * https://docs.oracle.com/javase/8/docs/api/java/time/ZoneId.html Example:
         * "UTC"
         */
        private String timeZone;

    }

    public static long getTime(String s, String locale, String format, ZoneId timeZone, TargetFormat targetFormat) throws ParseException {
        locale = StringUtils.defaultIfBlank(locale, Locale.US.toString());
        String finalLocale = locale;
        DateTimeFormatter dateTimeFormatter = sdfMap.computeIfAbsent(format + "|" + locale, f -> DateTimeFormatter.ofPattern(format, LocaleUtils.toLocale(finalLocale)));
        TemporalAccessor temporal = dateTimeFormatter.parseBest(s, LocalDateTime::from, ZonedDateTime::from, LocalDate::from);
        Instant instant;
        if (temporal instanceof LocalDateTime) {
            instant = ((LocalDateTime) temporal).atZone(Objects.nonNull(timeZone) ? timeZone : ZoneId.systemDefault()).toInstant();
        } else if (temporal instanceof ZonedDateTime) {
            instant = ZonedDateTime.from(temporal).toInstant();
        } else {
            instant = ((LocalDate) temporal).atStartOfDay().atZone(Objects.nonNull(timeZone) ? timeZone : ZoneId.systemDefault()).toInstant();
        }
        return TargetFormat.UNIX_MS.equals(targetFormat) ? instant.toEpochMilli() : instant.getEpochSecond();
    }
}